/*
 * Copyright Konstantin Triger <kostat@gmail.com> 
 * 
 * This file is part of Jaque - JAva QUEry library <http://code.google.com/p/jaque/>.
 * 
 * Jaque is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Jaque is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

package jaque;

import jaque.functions.*;
import jaque.misc.*;

import java.util.Collections;
import java.util.Iterator;

import static jaque.expressions.Functions.*;

/**
 * Provides a comprehensive set of methods receiving and returning object
 * sequences. Each method applies a business rule in a specific way and returns
 * a new object sequence for further processing.
 * 
 * A very useful application is feeding one function with output of another
 * function, i.e.<p/>
 * 
 * {@code select(where(...), ...). }
 * 
 * @author <a href="mailto://object_streaming@googlegroups.com">Konstantin
 *         Triger</a>
 */
public final class Query {

	private Query() {
	}

	/**
	 * Concatenates two object sequences. See also
	 * {@link Query#union(Iterable, Iterable)}.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @return the concatenated sequence.
	 */
	public static <T> Iterable<T> concat(final Iterable<T> first,
			final Iterable<T> second) {

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return concatIterator(first, second);
			}
		};
	}

	private static <T> Iterator<T> concatIterator(final Iterable<T> first,
			final Iterable<T> second) {
		return new BaseIterator<T>() {
			int _state;
			Iterator<T> _iter;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_iter = first.iterator();
					_state = 1;
				case 1:
					if (_iter.hasNext())
						return yield(_iter.next());

					_iter = second.iterator();
					_state = 2;
				case 2:
					if (_iter.hasNext())
						return yield(_iter.next());
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Returns the elements of the {@code source} sequence or the specified
	 * value in a singleton collection if the sequence is empty.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param source
	 *            the input object sequence.
	 * @param defaultValue
	 *            The value to return if the {@code source} sequence is empty.
	 * @return An {@code source} that contains {@code defaultValue} if source is
	 *         empty; otherwise, source.
	 */
	public static <T> Iterable<T> defaultIfEmpty(final Iterable<T> source,
			final T defaultValue) {

		if (defaultValue == null)
			throw new NullPointerException("value");

		return new Iterable<T>() {
			public final Iterator<T> iterator() {

				return defaultIfEmptyIterator(source, defaultValue);
			}
		};
	}

	private static <T> Iterator<T> defaultIfEmptyIterator(
			final Iterable<T> source, final T value) {
		return new BaseIterator<T>() {

			private int _state;
			private final Iterator<T> _iter = source.iterator();

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					if (_iter.hasNext()) {
						_state = 1;
						return yield(_iter.next());
					} else {
						_state = 2;
						return yield(value);
					}

				case 1:
					if (_iter.hasNext())
						return yield(_iter.next());
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Returns an empty object sequence of a specific type.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequence.
	 * @return an empty object sequence of a specific type.
	 */
	public static <T> Iterable<T> empty() {
		return Collections.<T> emptyList();
	}

	/**
	 * Correlates the objects of two sequences based on matching keys.
	 * 
	 * <pre>
	 * Object.equals()
	 * </pre>
	 * 
	 * is used to compare keys.
	 * 
	 * @param <T>
	 *            the type of the objects in the first (outer) sequence.
	 * @param <R>
	 *            the type of the objects in the second (inner) sequence.
	 * @param <Key>
	 *            The type of the keys returned by the key selector functions.
	 * @param <Result>
	 *            the type of the object in the resulting sequence.
	 * @param outer
	 *            the first sequence to join.
	 * @param outerSelector
	 *            a function to extract the join key from each object of the
	 *            first sequence.
	 * @param inner
	 *            the second sequence to join.
	 * @param innerSelector
	 *            a function to extract the join key from each object of the
	 *            second sequence.
	 * @param resultSelector
	 *            a function to create a result object from two matching
	 *            objects.
	 * @return an object sequence containing objects obtained by performing an
	 *         inner join on two sequences.
	 */
	public static <T, R, Key, Result> Iterable<Result> join(
			final Function<? extends Key, ? super T> outerSelector,
			final Function<? extends Key, ? super R> innerSelector,
			final Function2<Result, ? super T, ? super R> resultSelector,
			final Iterable<T> outer, final Iterable<R> inner) {
		return join(outerSelector, innerSelector, resultSelector, null, outer,
				inner);
	}

	/**
	 * Correlates the objects of two sequences based on matching keys. A
	 * specified comparator is used to compare keys.
	 * 
	 * @param <T>
	 *            the type of the objects in the first (outer) sequence.
	 * @param <R>
	 *            the type of the objects in the second (inner) sequence.
	 * @param <Key>
	 *            The type of the keys returned by the key selector functions.
	 * @param <Result>
	 *            the type of the object in the resulting sequence.
	 * @param outer
	 *            the first sequence to join.
	 * @param outerSelector
	 *            a function to extract the join key from each object of the
	 *            first sequence.
	 * @param inner
	 *            the second sequence to join.
	 * @param innerSelector
	 *            a function to extract the join key from each object of the
	 *            second sequence.
	 * @param resultSelector
	 *            a function to create a result object from two matching
	 *            objects.
	 * @param comparator
	 *            comparator is used to compare keys.
	 * @return an object sequence containing objects obtained by performing an
	 *         inner join on two sequences.
	 */
	public static <T, R, Key, Result> Iterable<Result> join(
			final Function<? extends Key, ? super T> outerSelector,
			final Function<? extends Key, ? super R> innerSelector,
			final Function2<Result, ? super T, ? super R> resultSelector,
			final java.util.Comparator<? super Key> comparator,
			final Iterable<T> outer, final Iterable<R> inner) {

		final java.util.Comparator<? super Key> comp = Utils
				.<java.util.Comparator<? super Key>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {
				return joinIterator(outerSelector, innerSelector,
						resultSelector, outer, inner, comp);
			}
		};
	}

	private static <Result, Key, T, R> Iterator<Result> joinIterator(
			final Function<? extends Key, ? super T> outerSelector,
			final Function<? extends Key, ? super R> innerSelector,
			final Function2<Result, ? super T, ? super R> resultSelector,
			final Iterable<T> outer, final Iterable<R> inner,
			final java.util.Comparator<? super Key> comp) {
		return new BaseIterator<Result>() {

			int _state;
			Iterator<T> _outer;
			Iterator<R> _inner;
			T _currentOuter;
			Key _outerKey;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_outer = outer.iterator();
					_state = 1;
				case 1:
					if (!_outer.hasNext())
						break;

					_currentOuter = _outer.next();
					_inner = inner.iterator();
					_outerKey = null;
					_state = 2;
				case 2:
					try {
						_outerKey = outerSelector.invoke(_currentOuter);
					} catch (RuntimeException re) {
						throw re;
					} catch (Throwable e) {
						throw new RuntimeException(e);
					}
					_state = 3;
				case 3:
					while (_inner.hasNext()) {
						R current = _inner.next();
						try {
							if (comp.compare(_outerKey, innerSelector
									.invoke(current)) == 0)
								return yield(resultSelector.invoke(
										_currentOuter, current));
						} catch (RuntimeException re) {
							throw re;
						} catch (Throwable e) {
							throw new RuntimeException(e);
						}
					}

					_state = 1;
					return hasNext();
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Transforms each object in the sequence into another object.
	 * 
	 * @param <T>
	 *            type of the objects in the input sequence.
	 * @param <Result>
	 *            type of the objects in the output sequence.
	 * @param source
	 *            the input object sequence.
	 * @param transform
	 *            a function to transform input object into output object.
	 * @return an object sequence containing transformed objects.
	 */
	public static <T, Result> Iterable<Result> select(
			final Function<? extends Result, ? super T> transform,
			final Iterable<T> source) {
		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {
				return selectIterator(transform, source);
			}
		};
	}

	private static <Result, T> Iterator<Result> selectIterator(
			final Function<? extends Result, ? super T> transform,
			final Iterable<T> source) {
		return new BaseIterator<Result>() {
			final private Iterator<T> _iter = source.iterator();

			// @Override
			public final boolean hasNext() {
				try {
					if (_iter.hasNext())
						return yield(transform.invoke(_iter.next()));
				} catch (RuntimeException re) {
					throw re;
				} catch (Throwable e) {
					throw new RuntimeException(e);
				}

				return yieldBreak;
			}
		};
	}

	public static <T, Result> Iterable<Result> selectMany(
			final Function<Iterable<Result>, ? super T> transform,
			final Iterable<T> source) {

		final Iterable<? extends Iterable<? extends Result>> multiSource = Query
				.select(transform, source);

		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {
				return selectManyIterator(multiSource);
			}
		};
	}

	private static <Result> Iterator<Result> selectManyIterator(
			final Iterable<? extends Iterable<? extends Result>> multiSource) {
		return new BaseIterator<Result>() {
			private int _state;
			private Iterator<? extends Iterable<? extends Result>> _outer;
			private Iterator<? extends Result> _inner;

			public final boolean hasNext() {
				switch (_state) {
				case 0:
					_outer = multiSource.iterator();
					_state = 1;
				case 1:
					if (!_outer.hasNext())
						return yieldBreak;

					_inner = _outer.next().iterator();
					_state = 2;
				default:
					if (!_inner.hasNext()) {
						_state = 1;
						return hasNext();
					}

					return yield(_inner.next());
				}
			}
		};
	}

	/**
	 * Recursively aggregates each object in the input sequence with the
	 * aggregated result of previous objects. e.g.
	 * 
	 * Sum(n objects) == Sum(n-1 objects) + n<sup>th</sup> object.
	 * 
	 * @param <T>
	 *            type of the objects in the input sequence.
	 * @param <Result>
	 *            type of the objects in the output sequence.
	 * @param source
	 *            the input object sequence.
	 * @param aggregator
	 *            a function to combine an aggregated result of the sequence
	 *            preceding current object with the current object.
	 * @return an object sequence containing aggregated results for each object
	 *         in the input sequence.
	 */
	public static <T, Result> Iterable<Result> aggregate(
			final Function2<? extends Result, ? super Result, ? super T> aggregator,
			final Iterable<T> source) {
		return new Iterable<Result>() {
			public final Iterator<Result> iterator() {
				return aggregateIterator(aggregator, source);
			}
		};

	}

	private static <T, Result> Iterator<Result> aggregateIterator(
			final Function2<? extends Result, ? super Result, ? super T> aggregator,
			final Iterable<T> source) {
		return new BaseIterator<Result>() {
			private final Iterator<T> _iter = source.iterator();

			// @Override
			public final boolean hasNext() {
				if (_iter.hasNext()) {
					try {
						return yield(aggregator.invoke(next(), _iter.next()));
					} catch (RuntimeException re) {
						throw re;
					} catch (Throwable e) {
						throw new RuntimeException(e);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Returns distinct elements from source sequence.
	 * 
	 * @param <T>
	 *            type of the objects in the input sequence.
	 * @param source
	 *            the input sequence.
	 * @return a sequence containing distinct elements.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> distinct(
			final Iterable<T> source) {
		return distinct(Comparator.<T> getComparableDefault(), source);
	}

	/**
	 * Returns distinct elements from source. A specified comparator is used to
	 * compare values.
	 * 
	 * @param <T>
	 *            type of the objects in the input sequence.
	 * @param source
	 *            the input sequence.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return a sequence containing distinct elements.
	 */
	public static <T> Iterable<T> distinct(
			final java.util.Comparator<? super T> comparator,
			final Iterable<T> source) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return distinctIterator(source, comp);
			}
		};
	}

	private static <T> Iterator<T> distinctIterator(final Iterable<T> source,
			final java.util.Comparator<? super T> comp) {
		return new BaseIterator<T>() {
			int _state;
			Iterator<T> _iter;
			java.util.Set<T> _set;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_set = new java.util.TreeSet<T>(comp);
					_iter = source.iterator();
					_state = 1;
				case 1:
					while (_iter.hasNext()) {
						T current = _iter.next();
						if (_set.add(current))
							return yield(current);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Produces the set difference of two input sequences.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @return a sequence that contains the set difference of two input
	 *         sequences.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> except(
			final Iterable<T> first, final Iterable<T> second) {
		return except(Comparator.<T> getComparableDefault(), first, second);
	}

	/**
	 * the set difference of two input sequences. A specified comparator is used
	 * to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @return a sequence that contains the set difference of two input
	 *         sequences.
	 */
	public static <T> Iterable<T> except(
			final java.util.Comparator<? super T> comparator,
			final Iterable<T> first, final Iterable<T> second) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return exceptIterator(first, second, comp);
			}
		};
	}

	private static <T> Iterator<T> exceptIterator(final Iterable<T> first,
			final Iterable<T> second, final java.util.Comparator<? super T> comp) {
		return new BaseIterator<T>() {
			int _state;
			Iterator<T> _iter;
			java.util.Collection<T> _set;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_set = new java.util.TreeSet<T>(comp);
					for (T t : second)
						_set.add(t);
					_iter = first.iterator();
					_state = 1;
				case 1:
					while (_iter.hasNext()) {
						T current = _iter.next();
						if (!_set.contains(current))
							return yield(current);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Produces the intersection of two sequences.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @return a sequence forming the intersection of two input sequences.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> intersect(
			final Iterable<T> first, final Iterable<T> second) {
		return intersect(Comparator.<T> getComparableDefault(), first, second);
	}

	/**
	 * Produces the intersection of two sequences. A specified comparator is
	 * used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return a sequence forming the intersection of two input sequences.
	 */
	public static <T> Iterable<T> intersect(
			final java.util.Comparator<? super T> comparator,
			final Iterable<T> first, final Iterable<T> second) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return intersectIterator(first, second, comp);
			}
		};
	}

	private static <T> Iterator<T> intersectIterator(final Iterable<T> first,
			final Iterable<T> second, final java.util.Comparator<? super T> comp) {
		return new BaseIterator<T>() {
			int _state;
			Iterator<T> _iter;
			java.util.Set<T> _set;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_set = new java.util.TreeSet<T>(comp);
					for (T t : second)
						_set.add(t);
					_iter = first.iterator();
					_state = 1;
				case 1:
					while (_iter.hasNext()) {
						T current = _iter.next();
						if (_set.remove(current))
							return yield(current);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Concatenates two sequence, excluding duplicates. A specified comparator
	 * is used to compare values. See also
	 * {@link Query#concat(Iterable, Iterable)}.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @return the concatenated sequence.
	 */
	public static <T extends Comparable<? super T>> Iterable<T> union(
			final Iterable<T> first, final Iterable<T> second) {
		return union(Comparator.<T> getComparableDefault(), first, second);
	}

	/**
	 * Concatenates two sequences, excluding duplicates. See also
	 * {@link Query#concat(Iterable, Iterable)}.
	 * 
	 * {@code Object.equals()}
	 * 
	 * is used to compare values.
	 * 
	 * @param <T>
	 *            the type of the objects in the sequences.
	 * @param first
	 *            the first sequence.
	 * @param second
	 *            the second sequence.
	 * @param comparator
	 *            comparator is used to compare values.
	 * @return the concatenated sequence.
	 */
	public static <T> Iterable<T> union(
			final java.util.Comparator<? super T> comparator,
			final Iterable<T> first, final Iterable<T> second) {

		final java.util.Comparator<? super T> comp = Utils
				.<java.util.Comparator<? super T>> coalesce(comparator,
						Comparator.getDefault());

		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return unionIterator(first, second, comp);
			}
		};
	}

	private static <T> Iterator<T> unionIterator(final Iterable<T> first,
			final Iterable<T> second, final java.util.Comparator<? super T> comp) {
		return new BaseIterator<T>() {
			int _state;
			Iterator<T> _iter;
			java.util.Set<T> _set;

			// @Override
			public final boolean hasNext() {

				switch (_state) {
				case 0:
					_iter = first.iterator();
					_set = new java.util.TreeSet<T>(comp);
					_state = 1;
				case 1:
					while (_iter.hasNext()) {
						T current = _iter.next();
						if (_set.add(current))
							return yield(current);
					}

					_iter = second.iterator();
					_state = 2;
				case 2:
					while (_iter.hasNext()) {
						T current = _iter.next();
						if (_set.add(current))
							return yield(current);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Generates a sequence consisting of {@code count} copies of the specified
	 * element.
	 * 
	 * @param <T>
	 *            the type of the object.
	 * @param element
	 *            an element to be repeated.
	 * @param count
	 *            number of times to repeat the element.
	 * @return a sequence consisting of {@code count} copies of the specified
	 *         element.
	 */
	public static <T> Iterable<T> repeat(final T element, final int count) {
		// return Collections.nCopies(count, element);
		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return repeatIterator(element, count);
			}
		};
	}

	private static <T> Iterator<T> repeatIterator(final T element,
			final int count) {
		return new BaseIterator<T>() {
			{
				yield(element);
			}

			int _count = count;

			// @Override
			public final boolean hasNext() {
				return _count-- > 0;
			}
		};
	}

	/**
	 * Bypasses a specified number of elements in a source and then returns the
	 * remaining ones.
	 * 
	 * @param <T>
	 *            the type of the objects in the source.
	 * @param source
	 *            the input sequence.
	 * @param count
	 *            number of elements to skip.
	 * @return a sequence consisting of remaining objects.
	 */
	public static <T> Iterable<T> skip(final int count, final Iterable<T> source) {
		return where(indexGreaterThanOrEqual(count), source);
	}

	/**
	 * Returns a specified number of elements from start of the object sequence.
	 * 
	 * @param <T>
	 *            the type of the object.
	 * @param source
	 *            the input object sequence.
	 * @param count
	 *            number of elements to return.
	 * @return filtered source.
	 */
	public static <T> Iterable<T> take(final int count, final Iterable<T> source) {
		return where(indexLessThan(count), source);
	}

	/**
	 * Returns elements from a source sequence as long as a specified condition
	 * is true.
	 * 
	 * @param <T>
	 *            the type of the objects in the source.
	 * @param source
	 *            the input sequence.
	 * @param predicate
	 *            A function to test each object in the source.
	 * @return filtered source.
	 */
	public static <T> Iterable<T> take(
			final Function<Boolean, ? super T> predicate,
			final Iterable<T> source) {
		return new Iterable<T>() {
			public final Iterator<T> iterator() {
				return takeIterator(predicate, source);
			}
		};
	}

	private static <E> Iterator<E> takeIterator(
			final Function<Boolean, ? super E> predicate,
			final Iterable<E> source) {
		return new BaseIterator<E>() {
			private final Iterator<? extends E> _iter = source.iterator();

			// @Override
			public final boolean hasNext() {

				if (_iter.hasNext()) {
					E current = _iter.next();
					try {
						if (predicate.invoke(current))
							return yield(current);
					} catch (RuntimeException re) {
						throw re;
					} catch (Throwable e) {
						throw new RuntimeException(e);
					}
				}

				return yieldBreak;
			}
		};
	}

	/**
	 * Filters the source according with the specified predicate.
	 * 
	 * @param <E>
	 *            the type of the objects in the source.
	 * @param source
	 *            the input sequence.
	 * @param predicate
	 *            A function to test each object in the source.
	 * @return filtered source.
	 */
	public static <E> Iterable<E> where(
			final Function<Boolean, ? super E> predicate,
			final Iterable<E> source) { // Comparable<T>
		return new Iterable<E>() {
			public final Iterator<E> iterator() {
				return whereIterator(predicate, source);
			}
		};
	}

	private static <E> Iterator<E> whereIterator(
			final Function<Boolean, ? super E> predicate,
			final Iterable<E> source) {
		return new BaseIterator<E>() {
			private final Iterator<? extends E> _iter = source.iterator();

			// @Override
			public final boolean hasNext() {

				while (_iter.hasNext()) {
					E current = _iter.next();
					try {
						if (predicate.invoke(current))
							return yield(current);
					} catch (RuntimeException re) {
						throw re;
					} catch (Throwable e) {
						throw new RuntimeException(e);
					}
				}

				return yieldBreak;
			}
		};
	}
}
